/*
    实现异步工作器
*/

#ifndef __L_LOOPER_H__
#define __L_LOOPER_H__

#include <iostream>
#include "buffer.hpp"
#include <thread>
#include <mutex>
#include <condition_variable>
#include <memory>
#include <functional>
#include <atomic>

namespace Log
{
    using Functor = std::function<void(Log::Buffer &)>;
    class AsyncLooper
    {
    public:
        using ptr = std::shared_ptr<AsyncLooper>;   //管理异步工作器的智能指针类型
        AsyncLooper(const Functor &callBack, BufferType looperType = BufferType::STATICSIZE)
        : _callBack(callBack),
        _looperType(looperType),
        _stop(false),
        _productorBuffer(_looperType),
        _consumerBuffer(_looperType),
        _thread(&AsyncLooper::threadEntry, this)
        {}

        ~AsyncLooper()
        {
            stop();
        }

        //停止异步工作器工作
        void stop()
        {
            // 在多线程的场景下可能会有多个线程修改_stop, 要么加锁, 要么定义为原子变量; 加锁可能会涉及到锁冲突, 所以定义为原子变量
            _stop = true;
            // 在停止日志工作器之前要将阻塞等待的工作线程唤醒后回收;
            _consumerCond.notify_all();
            //等待工作线程的退出
            _thread.join();
        }

        //往缓冲区中push数据
        void push(const char *data, size_t len)
        {
            //往缓冲区中push数据的两种情况:
            //1.动态扩容的缓冲区---不会有数据满了的情况,所以非常安全
            //2.固定大小的缓冲区---生产缓冲区中数据满了就要阻塞
            
            //多个生产者之间是互斥关系, 为了保证线程安全需要加锁
            std::unique_lock<std::mutex> lock(_mutex);

            //固定大小的缓冲区一般是用在业务场景中的不会因为没有内存而崩溃, 所以是安全的
            //动态增长的缓冲区是无限扩容的, 一般用于压力测试中, 如果没有内存了就会崩溃掉, 所以是不安全的
            if(_looperType == BufferType::STATICSIZE)
            {
                //业务场景下, 一般都是固定大小的缓冲区, 所以当缓冲区的剩余空间小于len时候就阻塞等待
                //C++中条件变量的成员函数wait传入的第二个参数为true的时候就不会阻塞等待, 为false的时候就会在条件变量上阻塞等待
                //当日志工作器没有停止并且缓冲区剩余空间的大小大于等于这次要写入的数据大小时就写入日志
                _productorCond.wait(lock, [&](){ return !_stop && _productorBuffer.writeAbleSize() >= len; });
            }

            //走到这里说明生产缓冲区中剩余空间足够这次数据的写入, 进行数据的写入
            _productorBuffer.push(data, len);
            // 数据写入到生产缓冲区后就唤醒工作线程对数据进行处理
            _consumerCond.notify_one();
        }

    private:
        //线程入口函数 --- 对消费缓冲区中的数据进行处理, 处理完毕后, 初始化缓冲区, 交换缓冲区
        void threadEntry()
        {
            //工作线程对消费者缓冲区中的数据持续的处理, 生产缓冲区中没有数据后就阻塞等待
            while(1)
            {
                // 为互斥锁设置一个声明周期, 当缓冲区交换完毕后就解锁(并不对数据的处理过程加锁保护, 因为工作线程是单线程)
                //  多线程的场景下, 交换缓冲区是需要加锁的, 不加锁有可能出现生产者线程正在往生产缓冲区中写数据时交换了缓冲区,
                // 会导致数据的不完整;
                // 但是工作线程处理数据是不需要加锁, 所以给加锁设置一个作用域;
                // 缓冲区交换结束后自动的解锁
                {

                    std::unique_lock<std::mutex> lock(_mutex);
                    // 刚启动异步工作器的时候, 消费缓冲区中的数据是空的, 如果生产缓冲区不是空的就进行缓冲区的交换, 否则工作线程就阻塞等待
                    // 另外异步工作器停止之前生产缓冲区中有数据,就需要将生产缓冲区中的数据交换到消费缓冲区后, 由工作线程落地后再停止;
                    _consumerCond.wait(lock, [&](){ return _stop || !_productorBuffer.empty(); });
                    if(_stop && _productorBuffer.empty())
                    {
                        break;
                    } 
                    //走到这里说明生产缓冲区中有数据就交换缓冲区
                    _consumerBuffer.swap(_productorBuffer);
                    //交换缓冲区后就可以唤醒生产者线程写入数据了
                    if(_looperType == BufferType::STATICSIZE)
                    {
                        //固定大小的缓冲区,当生产缓冲区满了生产者线程会阻塞, 才需要被唤醒, 动态扩容的缓冲区生产者线程不会阻塞,不需要唤醒
                        _productorCond.notify_all();
                    }
                }
                //调用回到函数, 对消费缓冲区中的数据实际落地
                _callBack(_consumerBuffer); //调用一次这个回调函数就将消费缓冲区中的数据全部落地
                //对消费缓冲区中的数据落地后, 初始化消费缓冲区
                _consumerBuffer.reset();
            }
        }

    private:
        Functor _callBack; // 具体对缓冲区中的数据进行处理的回调函数, 有异步工作器使用者传入;
    private:
        BufferType _looperType; //缓冲区的类型
        std::atomic<bool> _stop;  // 异步工作器是否停止
        Buffer _productorBuffer;  //生产者缓冲区
        Buffer _consumerBuffer; //消费者缓冲区
        std::mutex _mutex;  //互斥锁
        std::condition_variable _productorCond; //生产者条件变量
        std::condition_variable _consumerCond;  //消费者条件变量
        std::thread _thread; //异步工作器对应的工作线程
    };
}

#endif

